package cn.jly.flink.source2sink.mysql;

import cn.jly.flink.entity.Person;
import cn.jly.flink.utils.ConstantUtils;
import cn.jly.flink.utils.FlinkUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @PackageName cn.jly.flink.source2sink.custom
 * @ClassName MysqlSourceDemo
 * @Description 自定义source从mysql中读取数据
 * @Author 姬岚洋
 * @Date 2021/1/15 上午 10:11
 */
public class MysqlSourceDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = FlinkUtils.getStreamExecutionEnv();
        env.addSource(new MysqlSource())
                .print();

        FlinkUtils.executeStream(env, "MysqlSource");
    }

    static class MysqlSource extends RichSourceFunction<Person> {
        PreparedStatement preparedStatement;
        Connection connection;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // 获取连接
            connection = ConstantUtils.Mysql.getConnection();
            if (connection == null) {
                return;
            }
            String sql = "select * from tbl_person";
            preparedStatement = connection.prepareStatement(sql);
        }

        @Override
        public void run(SourceContext<Person> ctx) throws Exception {
            if (preparedStatement == null) {
                return;
            }
            final ResultSet resultSet = preparedStatement.executeQuery();
            while (resultSet.next()){
                final Person person = new Person();
                person.setId(resultSet.getLong("id"));
                person.setLastName(resultSet.getString("last_name").trim());
                person.setEmail(resultSet.getString("email").trim());
                person.setGender(resultSet.getInt("gender"));
                person.setAge(resultSet.getInt("age"));

                ctx.collect(person);
            }
        }

        @Override
        public void cancel() {
            // 释放资源
            ConstantUtils.Mysql.release(preparedStatement, connection);
        }
    }
}
